package org.gxlu.com.storm.wc;

import java.util.HashMap;
import java.util.Map;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("serial")
public class CountBolt  extends BaseRichBolt {

	private static Logger log = LoggerFactory.getLogger(CountBolt.class) ;
	
	private OutputCollector collector = null ;
	
	private Map<String,Integer> map = new HashMap<>();
	
	@SuppressWarnings("rawtypes")
	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector_) {
		  collector = collector_;
		
	}

	@Override
	public void execute(Tuple input) {
		String word = input.getString(0) ; // 得到上一个tuple中的第0列 
		Integer num = input.getInteger(1) ; //得到上一个tuple中的第1列 
		
		//或 
		//word = input.getStringByField("word"); // input.getFields().get(0)
		//num = input.getIntegerByField("num") ;		
		
		if(map.containsKey(word)){
			Integer count = map.get(word); // 取出上一个累加的值
			map.put(word, (count+num)) ; //  累加
		}else{
			map.put(word, num);
		}
		log.info("CountBolt-----threadId:"+Thread.currentThread().getId()+"map.hashcode:"+map.getClass().hashCode()+",map:"+map);
	   // System.out.println("threadId:"+Thread.currentThread().getId()+",map:"+map);
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		 
	}
 

}
